package com.atguigu.cep;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class FlinkCEPDemo2 {
    // 定义事件类
    public static class LoginEvent {
        public String userId;
        public String eventType; // "success" 或 "fail"
        public long timestamp;

        public LoginEvent() {}

        public LoginEvent(String userId, String eventType, long timestamp) {
            this.userId = userId;
            this.eventType = eventType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "LoginEvent{" +
                    "userId='" + userId + '\'' +
                    ", eventType='" + eventType + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

    // 定义匹配结果类
    public static class LoginPattern {
        public String userId;
        public String patternType; // "success_after_fails" 或 "timeout"
        public int failCount;

        public LoginPattern() {}

        public LoginPattern(String userId, String patternType, int failCount) {
            this.userId = userId;
            this.patternType = patternType;
            this.failCount = failCount;
        }

        @Override
        public String toString() {
            return "LoginPattern{" +
                    "userId='" + userId + '\'' +
                    ", patternType='" + patternType + '\'' +
                    ", failCount=" + failCount +
                    '}';
        }
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class TimeoutPattern {
        public String userId;
        public List<LoginEvent> timedOutEvents;
    }

    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 设置为单线程便于测试

        // 2. 创建模拟数据流
        DataStream<LoginEvent> loginEvents = env.fromElements(
                new LoginEvent("user1", "login", 1000L),
                // user1 没有后续成功事件，将超时
                new LoginEvent("user2", "login", 3000L),
                new LoginEvent("user2", "order", 4000L),

                // user3 只有一次失败，不满足模式，不会匹配也不会超时
                new LoginEvent("user3", "login", 6000L),

                // user4 有3次失败但没有成功事件，将超时
                new LoginEvent("user4", "login", 7000L),
                new LoginEvent("user4", "login", 18000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((event, timestamp) -> event.timestamp)
        );

        // 3. 按用户ID分组
        KeyedStream<LoginEvent, String> keyedLoginEvents = loginEvents.keyBy(event -> event.userId);

        // 4. 定义CEP模式：2次失败后跟1次成功，5秒内完成
        Pattern<LoginEvent, ?> pattern = Pattern.<LoginEvent>begin("login")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent event) {
                        return "login".equals(event.eventType);
                    }
                })
                .next("order")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent event) {
                        return "order".equals(event.eventType);
                    }
                })
                .within(Time.seconds(5)); // 5秒内完成

        // 5. 创建CEP模式流
        PatternStream<LoginEvent> patternStream = CEP.pattern(keyedLoginEvents, pattern);

        // 6. 定义侧输出标签
        final OutputTag<LoginPattern> normalMatchTag = new OutputTag<LoginPattern>("normal-match"){};
        final OutputTag<TimeoutPattern> timeoutTag = new OutputTag<TimeoutPattern>("timeout-match"){};

        // 7. 处理匹配结果
        SingleOutputStreamOperator<LoginPattern> mainDataStream =
                patternStream.process(new MyPatternProcessFunction(normalMatchTag, timeoutTag));

        // 8. 获取侧输出流
        DataStream<TimeoutPattern> timeoutStream = mainDataStream.getSideOutput(timeoutTag);

        // 9. 输出结果
        System.out.println("===== 正常匹配输出 (失败后成功登录) =====");
        mainDataStream.print();

        System.out.println("===== 超时匹配旁路输出 (未完成模式的事件序列) =====");
        timeoutStream.print();

        // 10. 执行作业
        env.execute("Flink CEP Timeout Pattern Detection");
    }

    @AllArgsConstructor
    @NoArgsConstructor
    public static class MyPatternProcessFunction extends
            PatternProcessFunction<LoginEvent, LoginPattern> implements TimedOutPartialMatchHandler<LoginEvent> {

        private OutputTag<LoginPattern> normalMatchTag;
        private OutputTag<TimeoutPattern> timeoutTag;

        @Override
        public void processMatch(Map<String, List<LoginEvent>> match, Context ctx, Collector<LoginPattern> out) throws Exception {
            // 正常匹配处理
            String userId = match.get("login").get(0).userId;
            out.collect(new LoginPattern(userId, "login_after_order", match.size()));
        }

        @Override
        public void processTimedOutMatch(Map<String, List<LoginEvent>> match, Context ctx) throws Exception {
            // 超时匹配处理
            String userId = match.get("login").get(0).userId;
            ctx.output(timeoutTag, new TimeoutPattern(userId, match.get("login")));
        }
    }
}
